package io.kiki.task.client.common.rpc.supports.handler;

import io.kiki.task.client.common.config.Properties;
import io.kiki.task.client.common.config.Properties.DispatcherThreadPool;
import io.kiki.task.client.common.rpc.supports.http.Request;
import io.kiki.task.client.common.rpc.supports.http.Response;
import io.kiki.task.common.core.enums.StatusEnum;
import io.kiki.task.common.core.model.HttpResponse;
import io.kiki.task.common.core.model.JobRequest;
import io.kiki.task.common.core.util.JsonUtil;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.handler.codec.http.*;
import io.netty.util.CharsetUtil;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.util.Map;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;


/**
 * @author huihui
 */
public class HttpServerHandler extends SimpleChannelInboundHandler<FullHttpRequest> {

    private final ThreadPoolExecutor dispatcherThreadPool;
    private final DispatcherRequestHandler dispatcherRequestHandler;

    public HttpServerHandler(DispatcherRequestHandler dispatcherRequestHandler, Properties properties) {
        this.dispatcherRequestHandler = dispatcherRequestHandler;

        // 获取线程池配置
        DispatcherThreadPool dispatcherThreadPool1 = properties.getDispatcherThreadPool();

        dispatcherThreadPool = new ThreadPoolExecutor(//
                dispatcherThreadPool1.getCorePoolSize(),//
                dispatcherThreadPool1.getMaximumPoolSize(),//
                dispatcherThreadPool1.getKeepAliveTime(),//
                dispatcherThreadPool1.getTimeUnit(), //
                new LinkedBlockingQueue<>(dispatcherThreadPool1.getQueueCapacity()),
                new CustomizableThreadFactory("netty-server-"));
    }

    @Override
    protected void channelRead0(ChannelHandlerContext channelHandlerContext, FullHttpRequest fullHttpRequest) {
        HttpHeaders httpHeaders = fullHttpRequest.headers();
        String content = fullHttpRequest.content().toString(CharsetUtil.UTF_8);
        String uri = fullHttpRequest.uri();
        HttpRequest httpRequest = HttpRequest.builder()
                .keepAlive(HttpUtil.isKeepAlive(fullHttpRequest))
                .uri(uri)
                .channelHandlerContext(channelHandlerContext)
                .httpMethod(fullHttpRequest.method())
                .httpHeaders(httpHeaders)
                .content(content)
                .response(new Response())
                .request(new Request(httpHeaders, uri))
                .build();

        // 执行任务
        dispatcherThreadPool.execute(() -> {
            HttpResponse httpResponse = null;
            try {
                httpResponse = dispatcherRequestHandler.dispatch(httpRequest);
            } catch (Exception e) {
                JobRequest retryRequest = JsonUtil.parseObject(content, JobRequest.class);
                httpResponse = new HttpResponse(StatusEnum.NO.getStatus(), e.getMessage(), null, retryRequest.getReqId());
            } finally {
                writeResponse(channelHandlerContext,
                        HttpUtil.isKeepAlive(fullHttpRequest),
                        httpRequest.getResponse(),
                        JsonUtil.toJsonString(httpResponse)
                );
            }
        });
    }

    private void writeResponse(ChannelHandlerContext ctx, boolean keepAlive, final Response httpResponse,
                               String responseJson) {
        // write response
        FullHttpResponse response = new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.OK,
                Unpooled.copiedBuffer(responseJson, CharsetUtil.UTF_8));
        response.headers().set(HttpHeaderNames.CONTENT_TYPE,
                HttpHeaderValues.APPLICATION_JSON);
        response.headers().set(HttpHeaderNames.CONTENT_LENGTH, response.content().readableBytes());
        if (keepAlive) {
            response.headers().set(HttpHeaderNames.CONNECTION, HttpHeaderValues.KEEP_ALIVE);
        }
        Map<String, Object> headers = httpResponse.getHttpHeaders();
        headers.forEach((key, value) -> response.headers().set(key, value));
        ctx.writeAndFlush(response);
    }

}
